package xyz.xuw.storm.workcount.bolt;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import xyz.xuw.storm.workcount.model.MyMessage;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by xuw on 2016/11/10.
 */
public class WordBolt extends BaseBasicBolt {
    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        MyMessage message = (MyMessage) tuple.getValues().get(0);
        String sentence = message.getWord();
        String[] words = sentence.split(" ");
        for (String word : words) {
            word = word.trim();
            if (!word.isEmpty()) {
                word = word.toLowerCase();
                basicOutputCollector.emit(new Values(new MyMessage(word)));
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word"));
    }
}
